/***************************************************************************

MapCG: MapReduce Framework for CPU & GPU

Copyright (C) 2010, Chuntao HONG (Chuntao.Hong@gmail.com).

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.

**************************************************************************/

#include <iostream>
#include <fstream>
#include <stdlib.h>
#include <string>
#include "time.h"
#include "math.h"
#include <vector>
#include <assert.h>
using namespace std;

#include "HMM/HMMMain.h"
#include "HMM/UtilLib/HMMCommUtil.h"
#include "HMM/UtilLib/HMMCommUtilTemplate.h"
#include "HMM/CPULib/HMMHashTableCPU.h"
using HMM_CPU::HashMultiMap;
using HMM_CPU::KeysIter;
using HMM_CPU::ValsIter;
#include "DS.h"

// #define TEST_DATA
#define TEST_SIZE 1000
#define TEST_SELECTIVITY 20

#define MAX_WORD_SIZE	255

int main(int argc, char **argv){   
	//----------------------------------------------
	//get parameters
	//----------------------------------------------
	string rFilename;
	string sFilename;

	// check input options are present
	if(!get_opt(argc,argv,"r1", rFilename) ||
		!get_opt(argc, argv, "r2", sFilename))
	{
			cout<<"usage: "<<argv[0]<<" -r1 relation1Filename -r2 relation2Filename " << endl;
			return 1;
	}

#ifndef TEST_DATA
	CRelation r;
	CRelation s;

	r.load(rFilename.c_str(), true);
	s.load(sFilename.c_str(), true);

	global_data_t global_data;
	unsigned int rSize = r.getCardinality();
	unsigned int sSize = s.getCardinality();
	global_data.r = r.getBATPointer();
	global_data.s = s.getBATPointer();

    // vector<unsigned int> rIndices;
    // vector<unsigned int> sIndices;
	unsigned int * rIndices = new unsigned int[rSize];	// offsets for the start of each tuple in the file
	unsigned int * sIndices = new unsigned int[sSize];

	for (unsigned int i = 0; i < rSize; ++i)
	{
		// rIndices.push_back(i);
        rIndices[i] = i;
	}

	for (unsigned int i = 0; i < sSize; ++i)
	{
		// sIndices.push_back(i);
        sIndices[i] = i;
	}
#else
    unsigned int rSize = TEST_SIZE;
    unsigned int sSize = TEST_SIZE;
    unsigned int sOffset = TEST_SIZE - (unsigned int)(TEST_SIZE * ((double)TEST_SELECTIVITY / 100)) + 1;
    SBUN r[rSize];
    SBUN s[sSize];
    unsigned int rIndices[rSize];
    unsigned int sIndices[sSize];

    for (int i = 0; i < TEST_SIZE; ++i)
    {
        r[i].key = i + 1;
        r[i].payload = i + 1;
        s[i].key = i + sOffset;
        s[i].payload = i + sOffset;
        rIndices[i] = i;
        sIndices[i] = i;
        /*printf("rKey: %u, rPayload: %u, sKey: %u, sPayload: %u, rIndex: %u, sIndex: %u\n",
                r[i].key,
                r[i].payload,
                s[i].key,
                s[i].payload,
                rIndices[i],
                sIndices[i]);*/
    }

    global_data_t global_data;
    global_data.r = r;
    global_data.s = s;
#endif

	// create MapReduce scheduler args
	HMMSchedulerSpec args(argc,argv);
	// args.input = &rIndices;
    args.input = rIndices;
	args.input_size=sizeof(int) * rSize;
	// args.input_2 = &sIndices;
    args.input_2 = sIndices;
	args.input_2_size = sizeof(int) * sSize;
	args.unit_size=sizeof(int);

	// args.PrintArgs();

	// create global data as part of scheduler args
	args.global_data=&global_data;

	// store the content and keyword in a global array
	DECLARE_GLOBAL_ARRAY(r, rSize * sizeof(SBUN));
	DECLARE_GLOBAL_ARRAY(s, sSize * sizeof(SBUN));

	// begin timing
	double t1=get_time();

	// create the map reduce scheduler
	HMMMainScheduler scheduler(args);
	
	// run map reduce
	scheduler.do_map_reduce();

	// end timing
	double t2=get_time();

	// output timing
	// cout << ", " << t2 - t1 << endl; << "Total time:     "<< t2-t1 << " seconds" << endl;

	// retrieve output
	OutputChunk output=scheduler.get_output();

	// output output
	HashMultiMap & hashMap = *(output.hashMultiMap);
	unsigned int numBuckets = hashMap.num_buckets;
	int numOutput = 0;
	int numJoined = 0;

	for (unsigned int i = 0; i < numBuckets; ++i)
	{
		KeysIter kIter = hashMap.getBucket(i);
		
		while (kIter.ptr != NULL)
		{
			ValsIter vIter = kIter.getValues();

			while (vIter.ptr != NULL)
			{
				if (numOutput < 10)
				{
					const void * val;
					unsigned int valSize;
					vIter.getValue(val, valSize);
					const SBUN * t = (const SBUN *)val;
					cout << "(" << t->key << ", " << t->payload << ")" << endl;
					++numOutput;
				}

				++numJoined;
				++vIter;
			}

			++kIter;
		}
	}

	cout << "Number of tuples joined: " << numJoined << endl;

	/*
	cout<<"number of output: "<<output.count<<endl;

	int joinCount = 0;
	for(int i=0;i<output.count; i++){
		unsigned int keyIdx=output.index[i].x;
		unsigned int keySize=output.index[i].y;
		unsigned int valIdx=output.index[i].z;
		unsigned int valSize = output.index[i].w;

		unsigned int key = *(int *)(output.keys + keyIdx);
		unsigned int * numOutput = (unsigned int*)(output.vals + valIdx);
		joinCount += *numOutput;
		outputTupleNode * out = (outputTupleNode *)(numOutput + sizeof(unsigned int));
		
		// cout << "Debug - no. of outputs in this array: " << *numOutput << endl;
		if (i < 10)
		{
			for (unsigned int j = 0; j < *numOutput; ++j)
			{
				//printf("(Key: %u, Value: %u, %u, %u, %u, %u, %u, Value Size: %u)\n", key, out[j].aId, out[j].bId, out[j].aKey, out[j].bKey, out[j].aData, out[j].bData, valSize);	
				printf("(Key: %u, Value: %u, %u, Value Size: %u)\n", key, out[j].aData, out[j].bData, valSize);	
			}
		}
	}

	cout << "Number of tuples joined: " << joinCount << endl;

	delete[] global_data.tuples;
	*/

	return 0;
}
